package com.corn.flink.lesson3;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;

import java.util.Properties;

/**
 * FlinkKafkaSinkOperatorDemo
 *
 * @author JimWu
 * @date 2023/2/28 17:16
 **/
public class FlinkKafkaSinkOperatorDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Person> ds = env.fromElements(
                new Person("alice", 15, "cq"),
                new Person("jack", 18, "cq"),
                new Person("rose", 20, "sh"),
                new Person("mick", 42, "gz"),
                new Person("white", 12, "cq"),
                new Person("maria", 32, "sz"),
                new Person("candy", 56, "bj"),
                new Person("jerry", 28, "wh")
        );

        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        ds.map(Person::toString)
                .addSink(new FlinkKafkaProducer<String>("my-person-topic", new SimpleStringSchema(), properties));

        env.execute();
    }
}
